package com.shujia.spark.mllib

import org.apache.spark.ml.linalg.{SparseVector, Vectors}
import org.apache.spark.sql.{DataFrame, Row, SaveMode, SparkSession}

object Demo5LoadImage {
  def main(args: Array[String]): Unit = {
    val spark: SparkSession = SparkSession
      .builder()
      .master("local[16]")
      .appName("point")
      .getOrCreate()
    import spark.implicits._
    import org.apache.spark.sql.functions._

    /**
     * 1、读取图片数据
     *
     */
    val imageDF: DataFrame = spark
      .read
      .format("image")
      .load("C:\\Users\\shujia\\Desktop\\mllib\\train")

    imageDF.printSchema()

    //取出图片路径和图片的数据
    val dataDF: DataFrame = imageDF.select($"image.origin" as "path", $"image.data" as "data")

    /**
     * 1、特征工程，将原始的图片的数据转换成向量
     */
    val featuresDF: DataFrame = dataDF.map {
      case Row(path: String, data: Array[Byte]) =>
        //处理数
        val xs: Array[Double] = data
          .map(byte => byte.toInt) //将二进制的数据转换成16进制
          //将白色的部分转换成1，将黑色部分转换成0
          .map((i: Int) => {
            if (i >= 0) {
              0.0
            } else {
              1.0
            }
          })
        //将特征转换成向量
        val features: SparseVector = Vectors.dense(xs).toSparse
        //取出图片名
        val name: String = path.split("/").last
        (name, features)
    }.toDF("name", "features")


    featuresDF.show(false)

    /**
     * 读取图片名和标记数据
     *
     */
    val nameAndLabelDF: DataFrame = spark
      .read
      .format("csv")
      .option("sep", " ")
      .schema("name STRING , label DOUBLE")
      .load("C:\\Users\\shujia\\Desktop\\mllib\\train.txt")


    val data: DataFrame = featuresDF
      //关联获取图片的数字
      .join(nameAndLabelDF.hint("broadcast"), "name")
      //取出目标值和特征向量
      .select($"label", $"features")

    data.show(false)

    /**
     * 将处理好的数据保存为svm格式
     */
    data
      .coalesce(10)
      .write
      .format("libsvm")
      .mode(SaveMode.Overwrite)
      .save("data/image_data")

  }

}
